// package collector provides a collector daemon for /debug/vars pages
// and graphing webservice.
package collector

import (
	"code.google.com/p/gomonitor/assets"
	cfg "code.google.com/p/gomonitor/config"
	"code.google.com/p/gomonitor/tsdb"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net/http"
	"os/exec"
	"path"
	"strconv"
	"strings"

	"time"
)

const layout = "01-02-2006.15:04-07:00"

func varPath(v string) []string {
	return strings.Split(v, ".")
}

var noSuchVariable = fmt.Errorf("No such variable")
var notIntVariable = fmt.Errorf("Not an int variable")

func fetchFloat64(p []string, m map[string]interface{}) (float64, error) {
	var item interface{} = m
	var ok bool
	for _, el := range p {
		item, ok = item.(map[string]interface{})[el]
		if !ok {
			return 0, noSuchVariable
		}
	}
	switch v := item.(type) {
	case int:
		return float64(v), nil
	case int32:
		return float64(v), nil
	case float32:
		return float64(v), nil
	case int64:
		return float64(v), nil
	case float64:
		return v, nil
	case string:
		return strconv.ParseFloat(v, 64)
	}
	return 0, notIntVariable
}

func gatherFromReader(rdr io.Reader, tdb *tsdb.Aggregator, scope string, varMap map[string]string) {
	//log.Printf("Gathering variables from scope %q", scope)
	vars := make(map[string]interface{})
	enc := json.NewDecoder(rdr)
	err := enc.Decode(&vars)
	if err != nil {
		log.Print("Error decoding vars", err)
		return
	}
	for v, _ := range varMap {
		m := tsdb.Metric{scope, v}
		//log.Printf("source: %q var %q m %v", scope, v, m)
		p := varPath(v)
		// fetch each value from the vars and store it in the database.
		// FEATURE handle strings?
		//log.Printf("Getting value for %q", m)
		val, err := fetchFloat64(p, vars)
		if err != nil {
			log.Printf("Error fetching %q from %q %v", v, scope, err)
		}
		ts := time.Now()
		log.Printf("Updating database for metric %v with value %v at timestamp %q, UnixNano %d", m, val, ts, ts.UnixNano())
		tdb.Add(m, float64(val), time.Now())
		if err != nil {
			log.Println("Error writing to tdb", err)
		}
	}
}

func gather(ss *cfg.ScrapeSpec, tdb *tsdb.Aggregator) {
	log.Println("Scraping vars from", ss.U.Url().String())
	r, err := http.Get(ss.U.Url().String())
	if r != nil {
		defer r.Body.Close()
	}
	if err != nil {
		log.Print("Error reading from", ss.U.Url().String(), err)
		return
	}
	gatherFromReader(r.Body, tdb, ss.Name, ss.Vars)
}

var query = []byte("get\n")

func queryExtern(c *Cmd, tdb *tsdb.Aggregator) {
	c.in.Write(query)
	gatherFromReader(c.out, tdb, c.c.Name, c.c.Vars)
}

type Cmd struct {
	c   *cfg.CmdSpec
	in  io.WriteCloser
	out io.ReadCloser
	err io.ReadCloser
}

func loop(specs []*cfg.ScrapeSpec, cs []*cfg.CmdSpec, tdb *tsdb.Aggregator, poll int) {
	interval := time.Second * time.Duration(poll)
	cmds := make(map[*Cmd]*exec.Cmd)
	for _, c := range cs {
		key := Cmd{c: c}
		cmd := exec.Command(c.Bin, c.Args...)
		var err error
		key.in, err = cmd.StdinPipe()
		if err != nil {
			log.Print("Could not get pipe to stdin for external ", err)
			return
		}
		key.out, err = cmd.StdoutPipe()
		if err != nil {
			log.Print("Could not get pipe to stdout for external ", err)
			return
		}
		key.err, err = cmd.StderrPipe()
		if err != nil {
			log.Print("Could not get pipe to stderr for external ", err)
			return
		}
		log.Printf("Starting command", c.Bin, c.Args)
		err = cmd.Start()
		if err == nil {
			cmds[&key] = cmd
		} else {
			log.Print(err.Error())
		}
	}
	for {
		// list of cfg.ScrapeSpecs to monitor
		ch := time.After(interval)
		for _, ss := range specs {
			// gather values
			gather(ss, tdb)
		}
		for k, _ := range cmds {
			queryExtern(k, tdb)
		}
		// wait for next interval
		<-ch
	}
}

const heartBeatMultiple = 3
const seasonalPeriod = 24

func viewHandler(w http.ResponseWriter, req *http.Request) {
	f, err := assets.Files.Open("index.html")
	if err != nil {
		log.Printf("error opening %q: %s", "index.html", err)
		http.NotFound(w, req)
		return
	}
	http.ServeContent(w, req, req.URL.Path, time.Time{}, f)
}

func dataToJson(rData []tsdb.SeriesResponse, tz *time.Location, labels ...string) []byte {
	data := [][]*float64{}
	var err string
	for _, r := range rData {
		if r.E != nil {
			err = r.E.Error()
		} else {
			t := float64(r.Val.Ts.In(tz).UnixNano())
			series := []*float64{&t}
			for _, v := range r.Val.Value {
				if v == nil {
					series = append(series, nil)
					continue
				}
				series = append(series, v)
			}
			data = append(data, series)
		}
	}
	bs, _ := json.Marshal(&struct {
		Err    string
		Data   [][]*float64
		Labels []string
	}{
		Err:    err,
		Data:   data,
		Labels: labels,
	})
	return bs
}

// reserved charactars ',' ':' ''
// /source:var,var2+source:var,var2/timeSpec
func feedHandler(w http.ResponseWriter, req *http.Request, tdb *tsdb.Aggregator) {
	feedReq, err := parseFeedUrl(req.URL.Path)
	if err != nil {
		log.Printf("Error parsing feed url %q", err)
		http.NotFound(w, req)
	}
	varAggNames := []varAgg{}
	metrics := []tsdb.Metric{}
	for _, fv := range feedReq.vars {
		varAggNames = append(varAggNames, fv.vars...)
		for _, n := range fv.vars {
			metrics = append(metrics, tsdb.Metric{fv.source, n.name})
		}
	}
	reqTz := feedReq.end.Location()
	labels := make([]string, 1, len(varAggNames)+1)
	labels[0] = "Time"
	// TODO(jwall): Use a parsed out roll duration.
	aggs := tsdb.AggSpec{Names: make([]string, len(metrics)), Span: feedReq.interval}
	for i, va := range varAggNames {
		nm := va.name
		if va.agg != "" {
			aggs.Names[i] = va.agg
			nm += ":" + va.agg
		} else {
			aggs.Names[i] = "Identity"
		}
		labels = append(labels, nm)
	}
	data := tdb.GetMultipleAggregations(metrics, aggs, feedReq.end, feedReq.duration)
	//log.Printf("XXX: Feed data: #v", data)
	w.Write(dataToJson(data, reqTz, labels...))
}

// Main starts the service with the provided Config, tsdb.Aggregator,
// a pollinterval, and the webservice host port to listen to.
func Main(config *cfg.Config, tdb *tsdb.Aggregator, poll int, listen string) {
	specs := config.Specs
	cs := config.Collectors
	log.Println("Config: %v", config)
	log.Println("Starting collection loop!!!")
	go loop(specs, cs, tdb, poll)
	// display graph page
	vf := func(w http.ResponseWriter, req *http.Request) {
		viewHandler(w, req)
	}
	ff := func(w http.ResponseWriter, req *http.Request) {
		feedHandler(w, req, tdb)
	}
	af := func(w http.ResponseWriter, req *http.Request) {
		p := path.Base(req.URL.Path)
		f, err := assets.Files.Open(p)
		if err != nil {
			log.Printf("error opening %q: %s", p, err)
			http.NotFound(w, req)
			return
		}
		http.ServeContent(w, req, req.URL.Path, time.Time{}, f)
	}
	http.HandleFunc("/view", vf)
	//        source varName
	// /feed/source/var/01-02-2006.15:04.CDT/30
	http.HandleFunc("/feed/", ff)
	http.HandleFunc("/assets/", af)
	log.Println("Starting http server on", listen)
	log.Fatal(http.ListenAndServe(listen, nil))
}
